package com.atguigu.realtime.dwd.db.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderPaySucDetail extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeOrderPaySucDetail().start(10016, 4, "dwd_trade_order_pay_suc_detail");
    }


    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {

        createTopicDb(groupId, tableEnv);

        fitlerPaymentTable(tableEnv);

        createDWDOrderDetail(tableEnv, groupId);

        createBasicDic(tableEnv);

        intervalJoin(tableEnv);

        Table resultTable = lookupJoin(tableEnv);

        createUpsertKafkaSink(tableEnv);

        resultTable.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS).execute();

    }

    private static void createUpsertKafkaSink(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table "+ Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS + "(\n" +
                "id STRING,\n" +
                "order_id STRING,\n" +
                "user_id STRING,\n" +
                "payment_type_code STRING,\n" +
                "payment_type_name STRING,\n" +
                "payment_time STRING,\n" +
                "sku_id STRING,\n" +
                "activity_id STRING,\n" +
                "activity_rule_id STRING,\n" +
                "coupon_id STRING,\n" +
                "sku_name STRING,\n" +
                "order_price STRING,\n" +
                "sku_num STRING,\n" +
                "split_total_amount STRING,\n" +
                "split_activity_amount STRING,\n" +
                "split_coupon_amount STRING,\n" +
                "province_id STRING,\n" +
                "ts bigint, \n" +
                "PRIMARY KEY (id) NOT ENFORCED \n" +
                ")" + SQLUtil.getUpsetKafkaSQL(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS));
    }

    private static Table lookupJoin(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select \n" +
                "id,\n" +
                "order_id,\n" +
                "user_id,\n" +
                "payment_type payment_type_code,\n" +
                "info.dic_name payment_type_name,\n" +
                "payment_time,\n" +
                "sku_id,\n" +
                "activity_id,\n" +
                "activity_rule_id,\n" +
                "coupon_id,\n" +
                "sku_name,\n" +
                "order_price,\n" +
                "sku_num,\n" +
                "split_total_amount,\n" +
                "split_activity_amount,\n" +
                "split_coupon_amount,\n" +
                "province_id,\n" +
                "ts\n" +
                "from pay_order p\n" +
                "left join base_dic  FOR SYSTEM_TIME AS OF p.proc_time as b \n" +
                "on p.payment_type = b.rowkey");
    }

    private static void intervalJoin(StreamTableEnvironment tableEnv) {
        Table payOrderTable = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "p.order_id,\n" +
                "p.user_id,\n" +
                "payment_type,\n" +
                "callback_time payment_time,\n" +
                "sku_id,\n" +
                "activity_id,\n" +
                "activity_rule_id,\n" +
                "coupon_id,\n" +
                "sku_name,\n" +
                "order_price,\n" +
                "sku_num,\n" +
                "split_total_amount,\n" +
                "split_activity_amount,\n" +
                "split_coupon_amount,\n" +
                "province_id,\n" +
                "p.ts,\n" +
                "p.proc_time\n" +
                "from payment p, order_detail od\n" +
                "where p.order_id = od.order_id\n" +
                "And p.row_time BETWEEN od.row_time - INTERVAL '15' MINUTE AND od.row_time + INTERVAL '5' SECOND");

        tableEnv.createTemporaryView("pay_order", payOrderTable);
    }

    private static void createDWDOrderDetail(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql("create table order_detail (\n" +
                "\tid STRING,\n" +
                "\torder_id STRING,\n" +
                "\tsku_id STRING,\n" +
                "\tactivity_id STRING,\n" +
                "\tactivity_rule_id STRING,\n" +
                "\tcoupon_id STRING,\n" +
                "\tsku_name STRING,\n" +
                "\torder_price STRING,\n" +
                "\tsku_num STRING,\n" +
                "\tcreate_time STRING,\n" +
                "\tsplit_total_amount STRING,\n" +
                "\tsplit_activity_amount STRING,\n" +
                "\tsplit_coupon_amount STRING,\n" +
                "\tuser_id STRING,\n" +
                "\tprovince_id STRING,\n" +
                "\tts bigint,\n" +
                " row_time as TO_TIMESTAMP_LTZ(ts * 1000,3), \n" +
                " WATERMARK FOR row_time AS row_time - INTERVAL '5' SECONDS \n" +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, groupId));
    }

    private static void fitlerPaymentTable(StreamTableEnvironment tableEnv) {
        Table paymentTable = tableEnv.sqlQuery("select \n" +
                "\t`data`['id'] id,\n" +
                "\t`data`['order_id'] order_id,\n" +
                "\t`data`['user_id'] user_id,\n" +
                "\t`data`['payment_type'] payment_type ,\n" +
                "\t`data`['total_amoun'] total_amount,\n" +
                "\t`data`['callback_time'] callback_time,\n" +
                "\tts,\n" +
                "\trow_time,\n" +
                "\tproc_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "and `table` = 'payment_info'\n" +
                "and `type` = 'update'\n" +
                "and `old`['payment_status'] is not null\n" +
                "and `data`['payment_status'] = '1602'");

        tableEnv.createTemporaryView("payment", paymentTable);
    }
}
